package com.exd.flink.stream.flow;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import com.exd.flink.common.HbaseTablesConstant;
import com.exd.flink.common.OutputVdataTagBase;
import com.exd.flink.common.PropertiesTool;
import com.exd.flink.domain.VDataBean;
import com.exd.flink.domain.VDataErrorBean;
import com.exd.flink.sink.clickhouse.ClickHouseSink;
import com.exd.flink.sink.hbase.HbaseSink;
import com.exd.flink.stream.function.VdataAnalyticalFunction;
import com.exd.flink.stream.function.VdataMergeFunction;

/**
 * 描述: 解析BdcmVdata数据
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class BdcmVdataStream {
	
	/**
	 * 解析BdcmVdata数据
	 */
	public static void analyticalData(DataStream<String> dataStreamValue) {
		// 筛选异常数据（不能解析）和正常的数据
		SingleOutputStreamOperator<VDataBean> process = dataStreamValue.process(new VdataAnalyticalFunction());
		// 获取解析正常的数据
		DataStream<VDataBean> jsondataToBeanDataOk = process.getSideOutput(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_OK);
		// 获取解析异常的数据
		DataStream<VDataErrorBean> jsondataToBeanDataError = process.getSideOutput(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_ERROR);
		// 合并数据
		SingleOutputStreamOperator<String> dataMergeStream = jsondataToBeanDataOk.keyBy(VDataBean::getKey).process(new VdataMergeFunction());
		// 获取全量整包数据
		DataStream<VDataBean> fullData = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_VDATA_PACKAGE_FULL);
		// 获取全量整包数据索引
		DataStream<VDataBean> fullDataIndex = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_VDATA_PACKAGE_FULL_INDEX);
		// 获取数采整包数据
		DataStream<VDataBean> miniData = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_VDATA_PACKAGE_MINI);
		// 获取事件整包数据
		DataStream<VDataBean> eventData = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_VDATA_PACKAGE_EVENT);
		// 获取切片数据
		DataStream<VDataBean> shardData = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_VDATA_SHARD);
		// 获取合并异常数据
		DataStream<VDataErrorBean> mergedataerror = dataMergeStream.getSideOutput(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_ERROR);
		// 合并数据流，合并异常的数据
		DataStream<VDataErrorBean> errorData = jsondataToBeanDataError.union(mergedataerror);
		// 判断是否进行持久化操作
		if(PropertiesTool.getParameterTool().getBoolean("exd.stroage.flag")) {
			//BDCM全量整包数据
			HbaseSink.vdataPackage(fullData, HbaseTablesConstant.BDCM_VDATA_PACKAGE_FULL);
			//BDCM数采整包数据
			HbaseSink.vdataPackage(miniData, HbaseTablesConstant.BDCM_VDATA_PACKAGE_MINI);
			//BDCM事件整包数据
			HbaseSink.vdataPackage(eventData, HbaseTablesConstant.BDCM_VDATA_PACKAGE_ENEVT);
			//BDCM分片数据
			HbaseSink.vdataShard(shardData, HbaseTablesConstant.BDCM_VDATA_SHARD);
			//BDCM异常数据
			HbaseSink.vdataError(errorData, HbaseTablesConstant.BDCM_VDATA_ERROR);
			//BDCM全量整包索引数据
			ClickHouseSink.inserBdcmVdataPackageFullIndex(fullDataIndex);
		}else {
			fullData.print("BDCM_PACKAGE_FULL:");
			miniData.print("BDCM_PACKAGE_MINI:");
			eventData.print("BDCM_PACKAGE_ENEVT:");
			shardData.print("BDCM_SHARD:");
			errorData.print("BDCM_ERROR:");
			fullDataIndex.print("BDCM_PACKAGE_FULL_INDEX:");	
		}
	}

}
